// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jhkim

#pragma once

#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
#include <boost/algorithm/string/replace.hpp>

#include "coin2/exchange/base/order/processor.h"
#include "coin2/exchange/sim/order_v1/backtesting_latency.h"
#include "coin2/exchange/sim/order_v1/matching_engine/matching_engine.h"
#include "coin2/strategy/linear_model/product_strategy.h"

#include "orio/record/simple.h"

namespace coin2::exchange::sim::order_v1 {

using coin2::exchange::order::AccountSimConfig;
using coin2::exchange::order::AccountInfoProto;
using coin2::exchange::order::OrderRecordProto;
using coin2::exchange::base::symbology::IProduct;
using coin::proto::OrderDuration;
using coin::proto::OrderDirection;

// for 1 account and 1 symbol
class BacktestingMatcher {
 public:
  BacktestingMatcher(
        int64_t matcher_id,
        const MarketExchangeApi& mea,
        const AccountSimConfig& config,
        const IProduct* product)
      : matcher_id(matcher_id),
        config_(config),
        holder_(new MatchingEngine(config)),
        mea_(mea),
        product_(product) {
    balance_.set_currency(product->base().currency());
    position_.set_symbol(product->absolute_norm());
    auto print_simbbo = getenv("PRINT_SIMBBO");
    print_simbbo_ = (print_simbbo && std::string(print_simbbo) == std::string("1"));
  }

  void LazyInitWriter() {
    if (config_.has_sim_result_filepath()) {
      std::string result_path_str = config_.sim_result_filepath();
      if (result_path_str.find("*") != std::string::npos) {
        // hack..
        boost::replace_all(result_path_str, "(.*)", product_->relative_norm());
      }
      std::experimental::filesystem::path result_path = result_path_str;
      writer_ = std::unique_ptr<orio::record::SimpleWriter>(orio::record::SimpleWriter::ToFile(
          result_path));
      LOG(INFO) << result_path.string();
      AccountInfoProto account_info;
      account_info.set_market_type(MarketType_Name(mea_.market));
      account_info.set_exchange(ExchangeType_Name(mea_.exchange));
      account_info.set_api(mea_.api);
      account_info.set_relative_norm(product_->relative_norm());
      account_info.set_reserve_position(config_.reserve_position());
      account_info.set_initial_position(config_.initial_position());
      account_info.set_start_timestamp(config_.start_timestamp());
      if (config_.has_end_timestamp()) {
        account_info.set_end_timestamp(config_.end_timestamp());
      } else {
        account_info.set_end_timestamp(0);
      }
      OrderRecordProto accinfo_row;
      accinfo_row.set_type(OrderRecordProto::ACCOUNT_INFO);
      *accinfo_row.mutable_account_info() = account_info;
      std::string buf;
      CHECK(accinfo_row.SerializeToString(&buf));
      writer_->Write(buf.data(), buf.size());
    }
  }

  int64_t account_id() const {
    return config_.sim_account_id();
  }

  void UpdateFeed(
      const presto::Symbol* symbol,
      const ::fastfeature::FeedMessage& feed);
  void SendOrder(
      int64_t order_id,
      std::optional<int64_t> last_time,
      const coin2::exchange::base::order::OrderSpec& spec,
      std::optional<double> submit_max_rate_used_ratio,
      std::optional<double> cancel_max_rate_used_ratio);
  void CancelOrder(int64_t order_id);

  ~BacktestingMatcher() {
    if (writer_) {
      if (config_.squeeze_stat()) {
        CHECK(config_.dump_fill_only());
        // force flush.
        LazyRecord(nullptr, true);
      }

      std::string buf;
      OrderRecordProto mutable_record;
      mutable_record.mutable_account_info()->set_end_position(holder_->GetPosition());
      mutable_record.mutable_account_info()->set_relative_norm(position_.symbol());
      mutable_record.mutable_account_info()->set_order_count(order_count);
      mutable_record.mutable_account_info()->set_order_qty(order_qty);
      mutable_record.mutable_account_info()->set_order_pq(order_pq);
      mutable_record.mutable_account_info()->set_fill_count(fill_count);
      mutable_record.mutable_account_info()->set_fill_qty(fill_qty);
      mutable_record.mutable_account_info()->set_fill_pq(fill_pq);
      mutable_record.mutable_account_info()->set_fill_qty_maker(fill_qty_maker);
      mutable_record.mutable_account_info()->set_fill_pq_maker(fill_pq_maker);
      mutable_record.mutable_account_info()->set_fill_qty_taker(fill_qty_taker);
      mutable_record.mutable_account_info()->set_fill_pq_taker(fill_pq_taker);
      mutable_record.mutable_account_info()->set_fill_qty_full(fill_qty_full);
      mutable_record.mutable_account_info()->set_fill_pq_full(fill_pq_full);
      mutable_record.mutable_account_info()->set_order_qty_filled(order_qty_filled);
      mutable_record.mutable_account_info()->set_order_pq_filled(order_pq_filled);
      mutable_record.mutable_account_info()->set_symbol_mtm_time(holder_->GetFetchedTime());
      mutable_record.mutable_account_info()->set_symbol_mtm_price(holder_->GetMtmPrice());
      if (first_sim_machine_ts_.has_value()) {
        mutable_record.mutable_account_info()->set_sim_elapsed_time(
            GetCurrentTimestamp() - *first_sim_machine_ts_);
        LOG(INFO) << "Sim elapsed time: " << mutable_record.account_info().sim_elapsed_time();
      }
      *mutable_record.mutable_account_info()->mutable_strat_aux_info() = config_.strat_aux_info();
      CHECK(mutable_record.SerializeToString(&buf));
      writer_->Write(buf.data(), buf.size());
    }
  }

  void WriteAndFlushRecords(bool first_write) {
    if (first_write) {
      first_sim_machine_ts_ = GetCurrentTimestamp();
    }
    if (!writer_) {
      LazyInitWriter();
    }
    const auto& records = holder_->GetOrderRecords();
    for (size_t i = 0; i < records.size(); ++i) {
      // OMIT other events due to pb size
      bool write_pb = false;
      if (records[i].type() == OrderRecordProto::ORDER_ACCEPTED) {
        if (config_.warmup_timestamp() <= records[i].time()) {
          order_count += 1;
          order_qty += records[i].order_qty();
          order_pq += records[i].order_qty() * records[i].order_price();
        }
      }
      switch (records[i].type()) {
        case OrderRecordProto::ORDER_ACCEPTED:
        case OrderRecordProto::ORDER_SUBMITTED:
        case OrderRecordProto::CANCEL_CONFIRMED:
        case OrderRecordProto::ORDER_ERROR:
          write_pb = !config_.dump_fill_only();
          break;
        case OrderRecordProto::ORDER_FILLED:
          if (config_.warmup_timestamp() <= records[i].time()) {
            fill_qty += records[i].fill_qty();
            fill_pq += records[i].fill_qty() * records[i].fill_price();
            if (records[i].fill_type() == OrderRecordProto::MAKER) {
              fill_qty_maker += records[i].fill_qty();
              fill_pq_maker += records[i].fill_qty() * records[i].fill_price();
            } else if (records[i].fill_type() == OrderRecordProto::TAKER) {
              fill_qty_taker += records[i].fill_qty();
              fill_pq_taker += records[i].fill_qty() * records[i].fill_price();
            }
            fill_count += 1 - fill_counts[records[i].order_id()];
            fill_counts[records[i].order_id()] = 1;
            fill_qtys[records[i].order_id()] += records[i].fill_qty();
            fill_pqs[records[i].order_id()] += records[i].fill_qty() * records[i].fill_price();
            order_qty_filled += records[i].order_qty() - order_qtys[records[i].order_id()];
            order_pq_filled += records[i].order_price() * records[i].order_qty() - order_pqs[records[i].order_id()];
            order_qtys[records[i].order_id()] = records[i].order_qty();
            order_pqs[records[i].order_id()] = records[i].order_price() * records[i].order_qty();
            if (fill_qtys[records[i].order_id()] == records[i].order_qty()) {
              fill_qty_full += fill_qtys[records[i].order_id()];
              fill_pq_full += fill_pqs[records[i].order_id()];
            }
          }
          write_pb = true;
        default:
          break;
      }
      if (write_pb) {
        if (config_.squeeze_stat()) {
          CHECK(config_.dump_fill_only());
          LazyRecord(&records[i], false);
        } else {
          WriteRecord(records[i]);
        }
      }
    }
    holder_->ClearOrderRecords();
    while (write_count > flush_frequency) {
      writer_->Flush();
      write_count = 0;
    }
  }

  void WriteRecord(const OrderRecordProto& record) {
    std::string buf;
    CHECK(record.SerializeToString(&buf));
    write_count += 1;
    writer_->Write(buf.data(), buf.size());
  }

  OrderRecordProto record_merges[4];

  void LazyRecord(const OrderRecordProto* record, bool force_flush) {
    if (record) {
      CHECK(!force_flush);
      int idx_r = 0;
      switch (record->sign()) {
        case 1:
          idx_r = 0;
          break;
        case -1:
          idx_r = 1;
          break;
        default:
          NOTREACHED() << record->sign();
      }
      switch (record->fill_type()) {
        case OrderRecordProto::MAKER:
          break;
        case OrderRecordProto::TAKER:
          idx_r += 2;
          break;
        default:
          NOTREACHED() << record->fill_type();
      }
      auto& record_merge = record_merges[idx_r];
      // initialization check
      const int64_t hour = 3'600'000'000'000LL;
      if ((record_merge.time() < config_.warmup_timestamp() &&
           config_.warmup_timestamp() <= record->time()) ||
          (record_merge.time() != 0LL &&
           record->time() / hour != record_merge.time() / hour)) {
        // if record_merge would span 0 o'clock, force flush it
        WriteRecord(record_merge);
        record_merge.Clear();
      }
      if (record_merge.time() == 0) {
        record_merge = *record;
      } else {
        double fill_pq_merge = record_merge.fill_qty() * record_merge.fill_price();
        double fill_pq_each = record->fill_qty() * record->fill_price();
        double fill_pq = fill_pq_merge + fill_pq_each;
        double fill_qty = record_merge.fill_qty() + record->fill_qty();
        record_merge.set_fill_qty(fill_qty);
        record_merge.set_fill_price(fill_pq / fill_qty);
        if (record->has_submit_max_rate_used_ratio()) {
          record_merge.set_submit_max_rate_used_ratio(record->submit_max_rate_used_ratio());
        }
        if (record->has_cancel_max_rate_used_ratio()) {
          record_merge.set_cancel_max_rate_used_ratio(record->cancel_max_rate_used_ratio());
        }
      }
      // flush every 1 min
      if (record->time() - record_merge.time() > 60e9) {
        WriteRecord(record_merge);
        record_merge.Clear();
      }
    } else if (force_flush) {
      for (auto& record_merge : record_merges) {
        WriteRecord(record_merge);
        record_merge.Clear();
      }
    }
  }

  const std::vector<OrderRecordProto>& GetOrderRecords() {
    return holder_->GetOrderRecords();
  }
  double GetNetPosition() const {
    return holder_->GetPosition() + config_.reserve_position();
  }
  double GetWorkingQty(int sign) const { return holder_->GetWorkingQty(sign); }
  double GetMtmPrice() const { return holder_->GetMtmPrice(); }
  coin::proto::ProductPosition* GetMutablePosition() { return &position_; }
  coin::proto::CurrencyBalance* GetMutableBalance() { return &balance_; }
  void UpdateTimeOnly(int64_t time) {
    // LOG(INFO) << "UpdateTimeOnly: " << time;
    holder_->UpdateTimeOnly(time);
  }

  int64_t matcher_id;
  int64_t write_count = 0;
  int64_t flush_frequency = 5000L;
  int64_t order_count = 0;
  int64_t fill_count = 0;
  double order_qty = 0;
  double order_pq = 0;
  double fill_qty = 0;
  double fill_pq = 0;
  double fill_qty_maker = 0;
  double fill_pq_maker = 0;
  double fill_qty_taker = 0;
  double fill_pq_taker = 0;
  std::unordered_map<int64_t, double> fill_counts;
  std::unordered_map<int64_t, double> fill_qtys;
  std::unordered_map<int64_t, double> fill_pqs;
  std::unordered_map<int64_t, double> order_qtys;
  std::unordered_map<int64_t, double> order_pqs;
  double order_qty_filled = 0;
  double order_pq_filled = 0;
  double fill_qty_full = 0;
  double fill_pq_full = 0;
  bool print_simbbo_ = false;

 private:
  AccountSimConfig config_;
  std::unique_ptr<orio::record::SimpleWriter> writer_;
  std::unique_ptr<MatchingEngine> holder_;
  coin::proto::ProductPosition position_;
  coin::proto::CurrencyBalance balance_;
  MarketExchangeApi mea_;
  const IProduct* product_;
  std::optional<int64_t> first_sim_machine_ts_;
};

}  // namespace coin2::exchange::sim::order_v1
